package com.deep.flink;

/**
 * @author sw
 * @create 2023-06-30 15:44
 */

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
//import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;


public class CommonApiTest {
    public static void main(String[] args) {
        //0.这里我们引入了一个“流式表环境”（StreamTableEnvironment），它是继承自TableEnvironment 的子接口。调用它的 create()方法，只需要直接将当前的流执行环境
        //（StreamExecutionEnvironment）传入，就可以创建出对应的流式表环境了。这也正是我们在上一节简单示例中使用的方式。
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //与1.0等效

        //1.定义环境配置来创建表执行环境
        //1.0基于blink版本planner进行流处理
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        //1.1基于老版本planner进行流处理 .useOldPlanner()

        //1.2基于老版本planner进行批处理
        //ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
        //BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(batchEnv);

        //1.3基于blink版本planner进行批处理
        EnvironmentSettings settings3 = EnvironmentSettings.newInstance()
                .inBatchMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tableEnv3 = TableEnvironment.create(settings3);


//        EnvironmentSettings settings = EnvironmentSettings.newInstance()
//                .inStreamingMode()
//                .useBlinkPlanner()
//                .build();
//        TableEnvironment tableEnv = TableEnvironment.create(settings);

        //2.创建表
        String createDDL = "CREATE TABLE clickTable (" +
                " user_name STRING," +
                " url STRING," +
                " ts BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/clicks.txt" +
                " 'format' = 'csv'" +
                ")";

        tableEnv.executeSql(createDDL);

        //创建用于输出的表
        String createOutDDL = "CREATE TABLE outTable (" +
                " url STRING," +
                " user_name STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'output'" +
                " 'format' = 'csv'" +
                ")";
        // 结果没用
        TableResult tableResult = tableEnv.executeSql(createOutDDL);

        tableEnv.useCatalog("custom_catalog");
        tableEnv.useDatabase("custom_database");



    }
}